#include "alp_global_vars.h"
#include "config_parser.h"

void recv_msg_recognize_type(int sd, short event, void * arg) {

	UNUSED_ARG(event);
	UNUSED_ARG(arg);

	char type;
	int status = recv(sd, &type, sizeof(char), MSG_PEEK);
	assert(status == sizeof(char));
	alp_interval_performance->num_interrupts++;

	switch (type) {

	case 'p':
		recv_proposal_message(sd);
		break;
	case 's':
		recv_short_notification_msg(sd);
		break;
	case 'i':
		recv_inst_msg(sd);
		break;
	case 'n':
		recv_inst_paxos_msg(sd);
		break;
	case 'm':
		recv_paxos_msg(sd);
		break;
	default: {
		printf("\nType of the received message is not recognized. TYPE: %c.\n", type);
		exit(1);
	}

	}

}

void recv_proposal_message(int sd) {
	proposer_msg p_msg;

	if (get_config_info()->self_acceptor_id == 0) { // Coordinator
		if (alp_interval_performance->next_instance_id < (INTERNAL_BUFFER_SIZE * get_config_info()->total_num_proposers) + alp_interval_performance->highest_deliverable_inst_id - 1) {
			int status = recv(sd, &p_msg, sizeof(p_msg), MSG_WAITALL);
			assert(status == sizeof(proposer_msg));
			alp_interval_performance->num_interrupts_new_msgs++;
			alp_interval_performance->num_interrupts_phase2b++;

			update_statistics(status, p_msg.highest_deliverable_inst_id);
			learner_deliver();
			generate_instance(&p_msg);

		} else {
			printf("i: %d hd: %d	BUF+hd-1: %d\n", alp_interval_performance->next_instance_id, alp_interval_performance->highest_deliverable_inst_id, (INTERNAL_BUFFER_SIZE * get_config_info()->total_num_proposers) + alp_interval_performance->highest_deliverable_inst_id - 1);
		}
	} else {
		int status = recv(sd, &p_msg, sizeof(p_msg), MSG_WAITALL);
		assert(status == sizeof(proposer_msg));
		alp_interval_performance->num_interrupts_new_msgs++;
		update_statistics(status, p_msg.highest_deliverable_inst_id);

		assert(get_config_info()->self_acceptor_id > 1); // acceptor 1 should never recv proposer msg.
		store_proposal(&p_msg);
	}

}

void recv_short_notification_msg(int sd) {

	short_dec_notification sd_msg;
	int status = recv(sd, &sd_msg, sizeof(short_dec_notification), MSG_WAITALL);
	assert(status == sizeof(short_dec_notification));
	update_statistics(status, sd_msg.highestDeliverable);

	alp_interval_performance->num_interrupts_phase2b_individual++;
	assert(sd_msg.instance_id != -1);

	if (get_config_info()->self_acceptor_id == 0) { // Coordinator
		// if there was the possibility of loss, this would have been considered. But because of TCP it is not.
	} else {
		//should not recv any
		printf("OOPS\n");
	}

}

void recv_inst_msg(int sd) {

	inst i_msg;
	int status = recv(sd, &i_msg, sizeof(inst), MSG_WAITALL);
	assert(status == sizeof(inst));
	update_statistics(status, i_msg.highestDeliverable);

	if (get_config_info()->self_acceptor_id == 0) { // Coordinator
		// Should not recv any
	} else {
		// Should not recv any
	}

}

void recv_inst_paxos_msg(int sd) {

	inst_paxos ip_msg;
	int status = recv(sd, &ip_msg, sizeof(inst_paxos), MSG_WAITALL);
	assert(status == sizeof(inst_paxos));
	update_statistics(status, ip_msg.inst_packet.highestDeliverable);

	if (get_config_info()->self_acceptor_id == 0) { // Coordinator
		//should not recv
	} else {
		a_handel_msg_inst_pack(&ip_msg);
		alp_interval_performance->num_interrupts_new_msgs++;
		alp_interval_performance->num_interrupts_phase2b++;
	}

}
void recv_paxos_msg(int sd) {
	paxos_short_not pd_msg;
	int status = recv(sd, &pd_msg, sizeof(paxos_short_not), MSG_WAITALL);
	assert(status == sizeof(paxos_short_not));
	alp_interval_performance->num_interrupts_phase2b_individual++;
	if (get_config_info()->self_acceptor_id == 0)
		printf("paxos\n");
	update_statistics(status, pd_msg.short_decision.highestDeliverable);
	a_handle_paxos_only(&pd_msg);

}

void update_statistics(int status, int hd) {

	alp_interval_performance->interval_bytes_received += status;
	double max_hd = max(alp_interval_performance->highest_deliverable_inst_id, hd);
	alp_interval_performance->highest_deliverable_inst_id = max_hd;

}

void store_proposal(proposer_msg *p_msg) {

	memcpy(&(p_msgs[p_msg->proposer_id][p_msg->location_index_in_proposer % INTERNAL_BUFFER_SIZE]), p_msg, sizeof(proposer_msg));
	p_msg->num_nodes_visited++;

	int status = send(get_config_info()->server_infos[get_config_info()->self_acceptor_id].successor_sid, p_msg, sizeof(proposer_msg), MSG_WAITALL);
	if (status != sizeof(proposer_msg)) {
		printf("NODE: %d SOCKET: %d, status: %dsizeofms: %d\n", get_config_info()->self_acceptor_id, get_config_info()->server_infos[get_config_info()->self_acceptor_id].successor_sid, status, sizeof(proposer_msg));
	}
	assert(status == sizeof(proposer_msg));
	alp_interval_performance->interval_bytes_sent += status;

}
double max(double a, double b) {
	if (a >= b)
		return a;
	else
		return b;
}

